package com.lrj.datastream.sink;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lrj.datastream.domain.ClickEvent;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * @author lrj
 * @date 2022/3/29 19:01
 */
public class TestRedisSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<ClickEvent> stream = env.fromElements(
                new ClickEvent("Mary", "./home", 1000L),
                new ClickEvent("Blob", "./home", 2000L),
                new ClickEvent("Mary", "./home", 3000L)
        );
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String host = parameterTool.get("host", "localhost");
        int port = parameterTool.getInt("port", 6379);
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder()
                .setHost(host)
                .setPort(port)
                .build();
        stream.addSink(new RedisSink<>(conf, new MyRedisMapper()));
        env.execute(TestRedisSink.class.getName());
    }

    private static class MyRedisMapper implements RedisMapper<ClickEvent> {
        @Override
        public RedisCommandDescription getCommandDescription() {
            // hset click k v
            return new RedisCommandDescription(RedisCommand.HSET, "click");
        }

        @Override
        public String getKeyFromData(ClickEvent clickEvent) {
            return clickEvent.getUser();
        }

        @Override
        public String getValueFromData(ClickEvent clickEvent) {
            ObjectMapper mapper = new ObjectMapper();
            try {
                return mapper.writeValueAsString(clickEvent);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            return "";
        }
    }
}
